/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.BindException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

import org.apache.jute.BinaryOutputArchive;
import org.apache.zookeeper.server.FinalRequestProcessor;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.RequestProcessor;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class has the control logic for the Leader.
 */
public class Leader {
    private static final Logger LOG = LoggerFactory.getLogger(Leader.class);
    
    static final private boolean nodelay = System.getProperty("leader.nodelay", "true").equals("true");
    static {
        LOG.info("TCP NoDelay set to: " + nodelay);
    }

    /**
     * 提议
     */
    static public class Proposal {
        // 集群间传递的包
        public QuorumPacket packet;

        // 接收到ack的机器sid集合
        public HashSet<Long> ackSet = new HashSet<>();

        // 请求
        public Request request;

        @Override
        public String toString() {
            return packet.getType() + ", " + packet.getZxid() + ", " + request;
        }
    }

    final LeaderZooKeeperServer zk;

    final QuorumPeer self;

    /**
     * 读取Learner消息的线程
     */
    LearnerCnxAcceptor cnxAcceptor;

    /**
     * learnerHandler集合
     */
    private final HashSet<LearnerHandler> learners = new HashSet<>();

    /**
     * Returns a copy of the current learner snapshot
     */
    public List<LearnerHandler> getLearners() {
        synchronized (learners) {
            return new ArrayList<LearnerHandler>(learners);
        }
    }

    /**
     * 参与者的LearnerHandler集合
     */
    private final HashSet<LearnerHandler> forwardingFollowers = new HashSet<>();
    
    /**
     * Returns a copy of the current forwarding follower snapshot
     */
    public List<LearnerHandler> getForwardingFollowers() {
        synchronized (forwardingFollowers) {
            return new ArrayList<LearnerHandler>(forwardingFollowers);
        }
    }

    private void addForwardingFollower(LearnerHandler lh) {
        synchronized (forwardingFollowers) {
            forwardingFollowers.add(lh);
        }
    }

    /**
     * 观察者LearnerHandler集合
     */
    private final HashSet<LearnerHandler> observingLearners = new HashSet<>();
        
    /**
     * Returns a copy of the current observer snapshot
     */
    public List<LearnerHandler> getObservingLearners() {
        synchronized (observingLearners) {
            return new ArrayList<LearnerHandler>(observingLearners);
        }
    }

    private void addObserverLearnerHandler(LearnerHandler lh) {
        synchronized (observingLearners) {
            observingLearners.add(lh);
        }
    }

    /**
     * 正在处理的同步(要等到过半ack才算完)
     */
    private final HashMap<Long,List<LearnerSyncRequest>> pendingSyncs = new HashMap<>();
    
    synchronized public int getNumPendingSyncs() {
        return pendingSyncs.size();
    }

    /**
     * 只不过是一个临时变量，并不是Follower count，看调用方
     */
    final AtomicLong followerCounter = new AtomicLong(-1);

    /**
     * Adds peer to the leader.
     * 
     * @param learner instance of learner handle
     */
    void addLearnerHandler(LearnerHandler learner) {
        synchronized (learners) {
            learners.add(learner);
        }
    }

    /**
     * Remove the learner from the learner list
     * 
     * @param peer
     */
    void removeLearnerHandler(LearnerHandler peer) {
        synchronized (forwardingFollowers) {
            forwardingFollowers.remove(peer);            
        }        
        synchronized (learners) {
            learners.remove(peer);
        }
        synchronized (observingLearners) {
            observingLearners.remove(peer);
        }
    }

    boolean isLearnerSynced(LearnerHandler peer){
        synchronized (forwardingFollowers) {
            return forwardingFollowers.contains(peer);
        }        
    }
    
    ServerSocket ss;

    Leader(QuorumPeer self,LeaderZooKeeperServer zk) throws IOException {
        this.self = self;
        try {
            ss = new ServerSocket();
            ss.setReuseAddress(true);
            // server.1 = 192.168.251.111:2888:3888
            // 2888: leader和follower互相之间进行通信的端口
            // 3888: 专门用来进行leader选举的端口
            ss.bind(new InetSocketAddress(self.getQuorumAddress().getPort()));
        } catch (BindException e) {
            LOG.error("Couldn't bind to port "
                    + self.getQuorumAddress().getPort(), e);
            throw e;
        }
        this.zk=zk;
    }

    /**
     * 数据同步型
     * leader -> Learner
     * 用于通知Learner服务器，Leader即将与其进行DIFF方式的数据同步
     */
    final static int DIFF = 13;
    
    /**
     * 数据同步型
     * Leader -> Learner
     * 用于触发Learner服务器进行内存数据库的回滚操作
     */
    final static int TRUNC = 14;
    
    /**
     * 数据同步型
     * Leader -> Learner
     * 用于通知Learner服务器，Leader即将与其进行SNAP方式的数据同步
     */
    final static int SNAP = 15;
    
    /**
     * 服务器初始化型
     * Observer -> Leader
     * 用于向Leader服务器注册自己，同时向Leader服务器表明当前Learner服务器是Observer, 消息中包含Observer的SID和ZXID
     */
    final static int OBSERVERINFO = 16;
    
    /**
     * 服务器初始化型
     * Leader -> Learner
     * Leader向learner发送阶段性的标识消息，leader与learner完成交互后，向learner发送 NEWLEADER消息，消息包含leder最新的ZXID
     */
    final static int NEWLEADER = 10;

    /**
     * 服务器初始化型
     * Follower -> Leader
     * 用于向leader注册自己，表名当前learner服务器是follower,消息中包含follower的sid和zxid
     */
    final static int FOLLOWERINFO = 11;

    /**
     * 数据同步型
     * Leader -> Learner
     * 用于告知Learner服务器已经完成了数据同步，可以开始对外提供服务
     */
    final static int UPTODATE = 12;

    /**
     * 服务器初始化型
     * leader -> learner
     * leader收到 OBSERVERINFO 和 FOLLOWERINFO后，会将自身的基本信息发送给这些learner,这个消息就是LEADERINFO，消息中包含leader的epoch值。
     */
    public static final int LEADERINFO = 17;

    /**
     * 服务器初始化型
     * learner -> leader
     * learner接受到leader发来的leaderinfo后，会将自己最新的zxid和epoch 以ACKEPOCH消息的形式发送给leader
     */
    public static final int ACKEPOCH = 18;
    
    /**
     * 请求处理型
     * learner -> leader
     * 请求转发消息，learner收到事务请求后，将该请求以REQUEST消息的形式转发给leader服务器处理
     */
    final static int REQUEST = 1;

    /**
     * 请求处理型
     * leader -> follower
     * zab协议中的提议。在处理事务请求时，leader服务器会将事务请求以PROPOSAL消息的形式创建投票发送给集群中的所有follower来进行事务日志的记录
     */
    public final static int PROPOSAL = 2;

    /**
     * 请求处理型
     * follower -> leader
     * follower接收到proposal消息后，会进行事务日志记录，如果完成记录后，就会以ACK消息的形式反馈给leader
     */
    final static int ACK = 3;

    /**
     * 请求处理型
     * leader -> follower
     * 用于通知集群中所有follower可以进行事务请求的提交了，leader在接收到过半的follower服务器发来的ack消息后，
     * 就进入事务请求的最终提交流程--生成COMMIT消息，告知所有的follower服务器进行事务请求的提交
     */
    final static int COMMIT = 4;

    /**
     * 会话管理型
     * leader -> learner
     * 同步learner上的客户端心跳检测，用于激活存活的客户端。
     * 客户端往往会随机地和任意一台服务器保持连接，因此leader无法直接接收到所有客户端的心跳检测， 需要委托给learner来保存这些客户端的心跳检测记录，
     * leader会定时地向learner发送PING消息，Learner在接收到PING消息后，会将这段时间内保持心跳的客户端列表， 同样以PING消息的形式反馈给Leader服务器，由leader负责
     */
    final static int PING = 5;

    /**
     * 会话管理型
     * learner -> leader
     * 检验会话是否有效，同时也会激活会话，通常发生在客户端重连的过程中，新的服务器需要向leader发送REVALIDATE消息以确定该会话是否已经超时
     */
    final static int REVALIDATE = 6;

    /**
     * 请求处理型
     * leader -> learner
     * 通知learner服务器已经完成了Sync操作
     */
    final static int SYNC = 7;
        
    /**
     * 请求处理型
     * leader -> Observer
     * 通知observer已经可以提交事务请求，同时还会在消息中携带事务请求的内容
     */
    final static int INFORM = 8;

    /**
     * 已经提出，还没有处理完的提议的map,key是zxid
     */
    ConcurrentMap<Long, Proposal> outstandingProposals = new ConcurrentHashMap<Long, Proposal>();

    /**
     * 即将生效的提议(已有过半确认)
     */
    ConcurrentLinkedQueue<Proposal> toBeApplied = new ConcurrentLinkedQueue<Proposal>();

    /**
     * newLeader的提议
     */
    Proposal newLeaderProposal = new Proposal();

    /**
     * 用于接收Learner的连接,启动LearnerHandler
     */
    class LearnerCnxAcceptor extends Thread{
        private volatile boolean stop = false;
        
        @Override
        public void run() {
            try {
                while (!stop) {
                    try{
                        //接收socket连接
                        Socket s = ss.accept(); // 阻塞住
                        s.setSoTimeout(self.tickTime * self.initLimit);
                        s.setTcpNoDelay(nodelay);
                        // 3. Leader服务器创建LearnerHandler(leader端)
                        // Leader接收到来自其他机器连接创建请求后，会创建一个LearnerHandler实例，
                        // 每个LearnerHandler实例都对应一个Leader与Learner服务器之间的连接，其负责Leader和Learner服务器之间几乎所有的消息通信和数据同步。
                        // 初次创建会收到其他服务器发送的OBSERVERINFO或是FOLLOWERINFO类型的消息。
                        LearnerHandler fh = new LearnerHandler(s, Leader.this); // 线程
                        fh.start();
                    } catch (SocketException e) {
                        if (stop) {
                            LOG.info("exception while shutting down acceptor: " + e);
                            stop = true;
                        } else {
                            throw e;
                        }
                    }
                }
            } catch (Exception e) {
                LOG.warn("Exception while accepting follower", e);
            }
        }
        
        public void halt() {
            stop = true;
        }
    }

    /**
     * leader状态详情
     */
    StateSummary leaderStateSummary;

    // 新的acceptEpoch号码,各端的max(lastAcceptedEpoch) + 1, 先默认-1
    long epoch = -1;
    // 是否在等待新的acceptEpoch号生成
    boolean waitingForNewEpoch = true;
    volatile boolean readyToStart = false;
    
    /**
     * 选举完leader之后，leader的入口函数。
     *
     * 1. 启动LearnerCnxAcceptor，利用LearnerHandler与各Learner进行IO
     * 2. Leader（以及LearnerHandler）调用getEpochToPropose，接收Leader（以及Learner向Leader注册时发送的LearnerInfo），更新epoch号,作为新的AcceptedEpoch和CurrentEpoch
     * 3. LearnerHandler把发送LEADERINFO包，把上面定的epoch也发送出去
     * 4. 调用waitForEpochAck,等待过半learner(以及leader),针对LEADERINFO返回ACKEPOCH包
     * 5. （省略:LearnHandler中间不断给各Learner同步数据）
     * 6. LearnerHandler发出NEWLEADER，Learner接收到，返回ACK
     * 7. LearnerHandler以及Leader调用waitForNewLeaderAck等待过半PARTICIPANT回复ACK
     * 8. 启动zkServer，不断完成ping，保持过半参与者同步
     * 
     * @throws IOException
     * @throws InterruptedException
     */
    void lead() throws IOException, InterruptedException {
        // 统计选举时间和注册JMX代码，忽略
        self.end_fle = System.currentTimeMillis();
        LOG.info("LEADING - LEADER ELECTION TOOK - " + (self.end_fle - self.start_fle));
        self.start_fle = 0; // 开始fast leader election时间
        self.end_fle = 0; // 结束时间

        zk.registerJMX(new LeaderBean(this, zk), self.jmxLocalPeerBean);

        try {
            self.tick = 0; // 初始tick为0，代表第几个tickTime
            // 1. 重新加载快照和事务日志数据
            zk.loadData();
            
            leaderStateSummary = new StateSummary(self.getCurrentEpoch(), zk.getLastProcessedZxid());
            // 2. Leader服务器启动Follower接收器LearnerCnxAcceptor(leader端)
            // 运行期间，Leader服务器需要和所有其余的服务器（统称为Learner）保持连接以确集群的机器存活情况，LearnerCnxAcceptor负责接收所有非Leader服务器的连接请求。用于集群间非选举通信
            cnxAcceptor = new LearnerCnxAcceptor(); // 等待learner的连接
            cnxAcceptor.start();
            
            readyToStart = true;
            // Leader解析Learner信息，计算新的epoch(leader端)
            long epoch = getEpochToPropose(self.getId(), self.getAcceptedEpoch());
            
            zk.setZxid(ZxidUtils.makeZxid(epoch, 0));
            
            synchronized(this){
                // 获取zxid
                lastProposed = zk.getZxid();
            }

            // 6. 发送Leader状态。(leader端)
            // 生成NEWLEADER包
            newLeaderProposal.packet = new QuorumPacket(NEWLEADER, zk.getZxid(), null, null);

            if ((newLeaderProposal.packet.getZxid() & 0xffffffffL) != 0) {
                LOG.info("NEWLEADER proposal has Zxid of " + Long.toHexString(newLeaderProposal.packet.getZxid()));
            }
            outstandingProposals.put(newLeaderProposal.packet.getZxid(), newLeaderProposal);

            newLeaderProposal.ackSet.add(self.getId());

            // 等待一段时间，收到过半的参与者返回的ACK
            waitForEpochAck(self.getId(), leaderStateSummary);
            self.setCurrentEpoch(epoch);

            while (!self.getQuorumVerifier().containsQuorum(newLeaderProposal.ackSet)){
                if (self.tick > self.initLimit) {
                    StringBuilder ackToString = new StringBuilder();
                    for(Long id : newLeaderProposal.ackSet) {
                        ackToString.append(id + ": ");
                    }
                    
                    shutdown("Waiting for a quorum of followers, only synced with: " + ackToString);
                    HashSet<Long> followerSet = new HashSet<>();

                    for(LearnerHandler f : getLearners()) {
                        followerSet.add(f.getSid());
                    }

                    if (self.getQuorumVerifier().containsQuorum(followerSet)) {
                        LOG.warn("Enough followers present. "+ "Perhaps the initTicks need to be increased.");
                    }
                    return;
                }
                Thread.sleep(self.tickTime);
                self.tick++;
            }
            
            String initialZxid = System.getProperty("zookeeper.testingonly.initialZxid");
            if (initialZxid != null) {
                long zxid = Long.parseLong(initialZxid);
                zk.setZxid((zk.getZxid() & 0xffffffff00000000L) | zxid);
            }

            if (!System.getProperty("zookeeper.leaderServes", "yes").equals("no")) {
                self.cnxnFactory.setZooKeeperServer(zk);
            }
            // 一个tick ping两次，
            boolean tickSkip = true;
    
            while (true) {
                // 每tickTime周期的一半
                Thread.sleep(self.tickTime / 2);
                if (!tickSkip) {
                    // 每两次代表一个tick
                    self.tick++;
                }
                HashSet<Long> syncedSet = new HashSet<>();

                syncedSet.add(self.getId());

                for (LearnerHandler f : getLearners()) {
                    // 如果是同步的，且是参与者
                    if (f.synced() && f.getLearnerType() == LearnerType.PARTICIPANT) {
                        syncedSet.add(f.getSid());
                    }
                    // 验证有没有LearnerHandler的proposal超时了没有处理
                    f.ping();
                }

              //需要验证时，集群验证过半验证失败
              if (!tickSkip && !self.getQuorumVerifier().containsQuorum(syncedSet)) {
                    shutdown("Only " + syncedSet.size() + " followers, need " + (self.getVotingView().size() / 2));
                    return;
              }
              // 翻转，进入下半个tickTime
              tickSkip = !tickSkip;
            }
        } finally {
            zk.unregisterJMX(this);
        }
    }

    boolean isShutdown;

    /**
     * Close down all the LearnerHandlers
     */
    void shutdown(String reason) {
        LOG.info("Shutting down");

        if (isShutdown) {
            return;
        }
        
        LOG.info("Shutdown called", new Exception("shutdown Leader! reason: " + reason));

        if (cnxAcceptor != null) {
            cnxAcceptor.halt();
        }
        
        self.cnxnFactory.setZooKeeperServer(null);
        try {
            ss.close();
        } catch (IOException e) {
            LOG.warn("Ignoring unexpected exception during close",e);
        }
        self.cnxnFactory.closeAll();
        if (zk != null) {
            zk.shutdown();
        }
        synchronized (learners) {
            for (Iterator<LearnerHandler> it = learners.iterator(); it.hasNext();) {
                LearnerHandler f = it.next();
                it.remove();
                f.shutdown();
            }
        }
        isShutdown = true;
    }

    /**
     * leader端：AckRequestProcessor调用了Leader#processAck
     * Follower端：SendAckRequestProcessor 发出ACK后，被LearnerHandler处理到，最终也是调用Leader#processAck，
     *
     * 收集ACK，如果通过过半验证了，就告诉所有参与者commit，告诉所有observer INFORM，然后调用commitProcessor逻辑
     *
     * @param zxid the zxid of the proposal sent out
     * @param followerAddr
     */
    synchronized public void processAck(long sid, long zxid, SocketAddress followerAddr) {
        // 没有处理当中的提议
        if (outstandingProposals.size() == 0) {
            return;
        }
        // 提议已经处理过了
        if (lastCommitted >= zxid) {
            return;
        }
        // 获取zxid对应的提议
        Proposal p = outstandingProposals.get(zxid);
        if (p == null) {
            LOG.warn("Trying to commit future proposal: zxid 0x{} from {}", Long.toHexString(zxid), followerAddr);
            return;
        }
        // 对应提议的ack集合添加sid记录
        p.ackSet.add(sid);
        // 判断一下，是否超过半数的follower返回了ack.
        // 瞬间挂掉了一个follower之后，你觉得是不是会有一定的影响呢？
        // 3个机器，half=1, 必须有2个ack, 只要leader和另一个follwer正常运行，对已经执行的操作没有影响，
        // 但是，对于新的操作会有影响，只有1个follower会返回ack，所以还是尽快恢复宕机的follower吧。
        if (self.getQuorumVerifier().containsQuorum(p.ackSet)){
            if (zxid != lastCommitted+1) {
                LOG.warn("Commiting zxid 0x{} from {} not first!", Long.toHexString(zxid), followerAddr);
                LOG.warn("First is 0x{}", Long.toHexString(lastCommitted + 1));
            }
            // 该proposal已经处理完了
            outstandingProposals.remove(zxid);
            if (p.request != null) {
                // 即将应用的队列 添加该proposal
                toBeApplied.add(p);
            }
            if ((zxid & 0xffffffffL) != 0) {
                if (p.request == null) {
                    LOG.warn("Going to commmit null request for proposal: {}", p);
                }
                // 提交，发给所有参与者
                commit(zxid);
                // 告诉所有观察者
                inform(p);
                // leader自己也提交
                zk.commitProcessor.commit(p.request);

                if(pendingSyncs.containsKey(zxid)){
                    for(LearnerSyncRequest r: pendingSyncs.remove(zxid)) {
                        // 发送同步请求给LearnerSyncRequest记录的server
                        sendSync(r);
                    }
                }
                return;
            } else {
                lastCommitted = zxid;
                LOG.info("Have quorum of supporters; starting up and setting last processed zxid: 0x{}", Long.toHexString(zk.getZxid()));
                zk.startup();
                zk.getZKDatabase().setlastProcessedZxid(zk.getZxid());
            }
        }
    }

    /**
     * 该处理器有一个toBeApplied队列，用来存储那些已经被CommitProcessor处理过的可被提交的Proposal。
     * 其会将这些请求交付给FinalRequestProcessor处理器处理，待其处理完后，再将其从toBeApplied队列中移除。
     *
     * 这个队列是在leader的processAck方法(即处理投票的方法)中加入数据的，存储的是leader已经进行commit，
     * 但是还没有进行内存数据结构操作的请求，主要是用于follower和leader的同步恢复。
     *
     * 只有leader角色才有，其他角色没有该处理器
     */
    static class ToBeAppliedRequestProcessor implements RequestProcessor {
        private RequestProcessor next;

        //注意是concurrent的
        private ConcurrentLinkedQueue<Proposal> toBeApplied;

        ToBeAppliedRequestProcessor(RequestProcessor next, ConcurrentLinkedQueue<Proposal> toBeApplied) {
            //下一个必须是FinalRequestProcessor
            if (!(next instanceof FinalRequestProcessor)) {
                throw new RuntimeException(ToBeAppliedRequestProcessor.class.getName() + " must be connected to " + FinalRequestProcessor.class.getName() + " not " + next.getClass().getName());
            }
            this.toBeApplied = toBeApplied;
            this.next = next;
        }

        /*
         * (non-Javadoc)
         * 
         * @see org.apache.zookeeper.server.RequestProcessor#processRequest(org.apache.zookeeper.server.Request)
         */
        @Override
        public void processRequest(Request request) throws RequestProcessorException {
            next.processRequest(request);
            //消息处理完后将对应的消息从toBeApplied队列中移除
            Proposal p = toBeApplied.peek();
            if (p != null && p.request != null && p.request.zxid == request.zxid) {
                toBeApplied.remove();
            }
        }

        /*
         * (non-Javadoc)
         * 
         * @see org.apache.zookeeper.server.RequestProcessor#shutdown()
         */
        public void shutdown() {
            LOG.info("Shutting down");
            next.shutdown();
        }
    }

    /**
     * QuorumPacket发给所有 参与者
     * 
     * @param qp the packet to be sent
     */
    void sendPacket(QuorumPacket qp) {
        synchronized (forwardingFollowers) {
            // 遍历所有的follower，
            for (LearnerHandler f : forwardingFollowers) {                
                f.queuePacket(qp);
            }
        }
    }
    
    /**
     * QuorumPacket发给所有 观察者
     */
    void sendObserverPacket(QuorumPacket qp) {        
        for (LearnerHandler f : getObservingLearners()) {
            f.queuePacket(qp);
        }
    }

    long lastCommitted = -1;

    /**
     * 更新lastCommitted，生成commit包，发给所有参与者
     * @param zxid
     */
    public void commit(long zxid) {
        synchronized(this){
            // 更新最近commit的zxid
            lastCommitted = zxid;
        }
        // 生成commit消息
        QuorumPacket qp = new QuorumPacket(Leader.COMMIT, zxid, null, null);
        // 发给所有参与者
        sendPacket(qp);
    }
    
    /**
     * 生成inform包,告诉所有observer
     * @param proposal
     */
    public void inform(Proposal proposal) {   
        QuorumPacket qp = new QuorumPacket(Leader.INFORM, proposal.request.zxid,  proposal.packet.getData(), null);
        // 告诉所有observer
        sendObserverPacket(qp);
    }

    long lastProposed;

    
    /**
     * Returns the current epoch of the leader.
     * 
     * @return
     */
    public long getEpoch(){
        return ZxidUtils.getEpochFromZxid(lastProposed);
    }

    /**
     * Xid回滚异常，一般用于Xid低32位满了的时候用(高32位是epoch号码)
     */
    public static class XidRolloverException extends Exception {
        public XidRolloverException(String message) {
            super(message);
        }
    }

    /**
     * 根据Request产生提议,发给所有参与者
     * 
     * @param request
     * @return the proposal that is queued to send to all the members
     */
    public Proposal propose(Request request) throws XidRolloverException {
        // 如果zxid低的32位已经满了，为了不溢出，就关闭掉，重新选举
        if ((request.zxid & 0xffffffffL) == 0xffffffffL) {
            String msg = "zxid lower 32 bits have rolled over, forcing re-election, and therefore new epoch start";
            shutdown(msg);
            //xid回滚异常
            throw new XidRolloverException(msg);
        }

        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
        try {
            request.hdr.serialize(boa, "hdr");
            if (request.txn != null) {
                request.txn.serialize(boa, "txn");
            }
            baos.close();
        } catch (IOException e) {
            LOG.warn("This really should be impossible", e);
        }
        // 产生提议
        QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid,  baos.toByteArray(), null);
        
        Proposal p = new Proposal();
        p.packet = pp;
        p.request = request;
        synchronized (this) {
            // 更新最近一次提议的zxid
            lastProposed = p.packet.getZxid();
            // 出现了一个尚未处理的提议
            outstandingProposals.put(lastProposed, p);
            // 核心方法：提议发给所有参与者
            sendPacket(pp);
        }
        return p;
    }
            
    /**
     * 处理同步请求,注意上锁了
     * @param r the request
     */
    
    synchronized public void processSync(LearnerSyncRequest r){
        // 没有正在处理的提议
        if(outstandingProposals.isEmpty()){
            //发送同步请求给LearnerSyncRequest记录的server
            sendSync(r);
        } else {
            // 把当时最新的lastProposed记录下来
            List<LearnerSyncRequest> l = pendingSyncs.get(lastProposed);
            if (l == null) {
                l = new ArrayList<LearnerSyncRequest>();
            }
            // lastProposed对应的记录添加进当前的request,这个列表的同步都是到lastProposed这个位置
            l.add(r);
            // 放入map，等到这个提议通过
            pendingSyncs.put(lastProposed, l);
        }
    }
        
    /**
     * 发送Sync请求给合适的server(LearnerSyncRequest记录的)
     * @param r
     */
    public void sendSync(LearnerSyncRequest r){
        QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
        r.fh.queuePacket(qp);
    }
                
    /**
     * learner同步的时候，同步方式是和leader的commitLog相关的，在此期间，记录在内存中的提议，以及即将生效的提议也要告诉给learner
     *  让leader知道Follower在进行同步,另外看zk当前有没有新的提议,或者同步的信息发送过去的
     * @param handler handler of the follower
     * @return last proposed zxid
     */
    synchronized public long startForwarding(LearnerHandler handler, long lastSeenZxid) {
        // 如果自己的zxid比发给learner的zxid大
        if (lastProposed > lastSeenZxid) {
            for (Proposal p : toBeApplied) {
                if (p.packet.getZxid() <= lastSeenZxid) {
                    // 即将生效的zxid，对应learner已经有记录了
                    continue;
                }
                handler.queuePacket(p.packet);
                // 添加commit
                QuorumPacket qp = new QuorumPacket(Leader.COMMIT, p.packet.getZxid(), null, null);
                // 加入handler的发送队列
                handler.queuePacket(qp);
            }
            List<Long>zxids = new ArrayList<>(outstandingProposals.keySet());
            Collections.sort(zxids);
            for (Long zxid: zxids) {
                if (zxid <= lastSeenZxid) {
                    continue;
                }
                handler.queuePacket(outstandingProposals.get(zxid).packet);
            }
        }
        //如果是参与者，就加到 参与者集合中
        if (handler.getLearnerType() == LearnerType.PARTICIPANT) {
            //加入 参与者 集合中
            addForwardingFollower(handler);
        } else {
            //否则加入到观察者 集合中
            addObserverLearnerHandler(handler);
        }
                
        return lastProposed;
    }

    /**
     * 连接上leader的sid集合
     */
    private HashSet<Long> connectingFollowers = new HashSet<>();

    /**
     * 被Leader和LearnerHandler调用，即获取集群最大的lastAcceptedEpoch，+1用于设置新的acceptedEpoch
     * @param sid
     * @param lastAcceptedEpoch
     * @return
     * @throws InterruptedException
     * @throws IOException
     */
    public long getEpochToPropose(long sid, long lastAcceptedEpoch) throws InterruptedException, IOException {
        synchronized(connectingFollowers) {
            // 如果还在等待new epoch
            if (!waitingForNewEpoch) {
                return epoch;
            }

            // 选出leader以及learner中最大一个lastAcceptedEpoch
            if (lastAcceptedEpoch >= epoch) {
                // 更新自己的epoch为最大的+1,
                epoch = lastAcceptedEpoch+1;
            }

            // 连接的Learner添加记录
            connectingFollowers.add(sid);
            QuorumVerifier verifier = self.getQuorumVerifier();

            // 如果自己也连接上,并且已经有过半机器连接上
            if (connectingFollowers.contains(self.getId()) &&  verifier.containsQuorum(connectingFollowers)) {
                waitingForNewEpoch = false; //不用再等了
                self.setAcceptedEpoch(epoch); //设置acceptedEpoch
                connectingFollowers.notifyAll();
            } else {
                long start = System.currentTimeMillis();
                long cur = start;
                long end = start + self.getInitLimit()*self.getTickTime();

                // 如果已经有机器连接上leader了，那么最多等待一段时间直到其他机器通过 过半验证
                while(waitingForNewEpoch && cur < end) {
                    connectingFollowers.wait(end - cur);
                    cur = System.currentTimeMillis();
                }

                if (waitingForNewEpoch) {
                    throw new InterruptedException("Timeout while waiting for epoch from quorum");        
                }
            }
            return epoch;
        }
    }

    /**
     * 针对LEADERINFO回复ACKEPOCH的集合
     */
    private HashSet<Long> electingFollowers = new HashSet<Long>();
    /**
     * 是否过半机器注册成功
     */
    private boolean electionFinished = false;

    /**
     * 验证leader的StateSummary是最新的,
     * 等待过半机器(Learner和leader)针对Leader发出的LEADERINFO回复ACKEPOCH
     * @param id
     * @param ss
     * @throws IOException
     * @throws InterruptedException
     */
    public void waitForEpochAck(long id, StateSummary ss) throws IOException, InterruptedException {
        synchronized(electingFollowers) {
            // 如果已经过半返回了
            if (electionFinished) {
                return;
            }
            if (ss.getCurrentEpoch() != -1) {
                //如果有currentEpoch，zxid比Leader自己的还要新
                if (ss.isMoreRecentThan(leaderStateSummary)) {
                    throw new IOException("Follower is ahead of the leader, leader summary: " 
                                                    + leaderStateSummary.getCurrentEpoch() + " (current epoch), "
                                                    + leaderStateSummary.getLastZxid() + " (last zxid)");
                }
                electingFollowers.add(id);
            }
            QuorumVerifier verifier = self.getQuorumVerifier();
            // 过半验证通过
            if (electingFollowers.contains(self.getId()) && verifier.containsQuorum(electingFollowers)) {
                electionFinished = true;
                electingFollowers.notifyAll();
            } else {                
                long start = System.currentTimeMillis();
                long cur = start;
                long end = start + self.getInitLimit() * self.getTickTime();
                // 等待一段时间
                while(!electionFinished && cur < end) {
                    electingFollowers.wait(end - cur);
                    cur = System.currentTimeMillis();
                }
                // 等待结束了还没有过半验证通过
                if (!electionFinished) {
                    throw new InterruptedException("Timeout while waiting for epoch to be acked by quorum");
                }
            }
        }
    }

    /**
     * Get string representation of a given packet type
     * @param packetType
     * @return string representing the packet type
     */
    public static String getPacketType(int packetType) {
        switch (packetType) {
        case DIFF:
            return "DIFF";
        case TRUNC:
            return "TRUNC";
        case SNAP:
            return "SNAP";
        case OBSERVERINFO:
            return "OBSERVERINFO";
        case NEWLEADER:
            return "NEWLEADER";
        case FOLLOWERINFO:
            return "FOLLOWERINFO";
        case UPTODATE:
            return "UPTODATE";
        case LEADERINFO:
            return "LEADERINFO";
        case ACKEPOCH:
            return "ACKEPOCH";
        case REQUEST:
            return "REQUEST";
        case PROPOSAL:
            return "PROPOSAL";
        case ACK:
            return "ACK";
        case COMMIT:
            return "COMMIT";
        case PING:
            return "PING";
        case REVALIDATE:
            return "REVALIDATE";
        case SYNC:
            return "SYNC";
        case INFORM:
            return "INFORM";
        default:
            return "UNKNOWN";
        }
    }
}
